-- **************************************************************************
--    Project Name:   dwd层-网点中心综合账单 清洗
--    Job Name:       spmi_dwd.dwd_spmn_apack_bill_dt（分区字段：账单产生时间）
--    Description :   从ods层spmi_union_bill表取每天T-1增量更新的数据，T-1 更新的数据划分为三部分，A： 最近30天账单产生的数据， B：最近31-60天账单产生的数据，C：60天之前账单产生的数据，
--                    A部分数据和 spmi_dwd.dwd_spmn_apack_bill_dt 最近30天基于运单号去重，然后回刷回0-30天分区；
--                    B部分数据和  spmi_dwd.dwd_spmn_apack_bill_dt 最近30-60天数据基于运单号去重，然后回刷回31-60天分区
--                    C部分数据直接丢弃，spmi_dwd.dwd_spmn_apack_bill_dt 60天之前的 数据不予更新
--    Author :        songjun
--    date：          2022/5/23
-- **************************************************************************
--    modify by songjun 2022/05/23 上线
-- **************************************************************************

SET hive.merge.mapfiles = true; --  默认true，在map-only任务结束时合并小文件
SET hive.merge.mapredfiles = true; -- 默认false，在map-reduce任务结束时合并小文件
SET hive.merge.size.per.task = 25610001000; -- 默认256M
SET hive.merge.smallfiles.avgsize = 16777216; -- 当输出文件的平均大小小于16m该值时，启动一个独立的map-reduce任务进行文件merge

-- 取ods层昨天数据
with tmp_spmi_union_bill_incr as (
    select
        id
        ,waybill_no
        ,settle_network_name
        ,settle_network_code
        ,settle_network_id
        ,financial_center_name
        ,financial_center_code
        ,financial_center_id
        ,waybill_num
        ,waybill_piece
        ,cost_weight
        ,settle_weight
        ,transfer_cost
        ,operation_cost
        ,piece_fee_cost
        ,distribu_depart_cost
        ,distribu_opera_cost
        ,is_package_cost
        ,product_type_id
        ,product_type_name
        ,goods_type_id
        ,goods_type_name
        ,transport_type_id
        ,transport_type_name
        ,customer_code
        ,customer_name
        ,begin_id
        ,begin_name
        ,sender_province_id
        ,sender_province_name
        ,destination_id
        ,destination_name
        ,receiver_province_id
        ,receiver_province_name
        ,waybill_time
        ,sign_time
        ,send_financial_center_id
        ,send_financial_center_name
        ,send_financial_center_code
        ,send_network_id
        ,send_network_code
        ,send_network_name
        ,sign_financial_center_id
        ,sign_financial_center_code
        ,sign_financial_center_name
        ,sign_network_id
        ,sign_network_code
        ,sign_network_name
        ,operation_network_code
        ,operation_network_name
        ,operation_financial_center_code
        ,operation_financial_center_name
        ,is_sign
        ,is_bad_piece
        ,is_refund
        ,remarks
        ,version
        ,bill_generation_date
        ,create_time
        ,update_time
        ,last_update_time_sync
        ,trace_id
        ,customer_fee
        ,distribu_version
        ,waybill_weight
        ,package_charge_weight_source
        ,center_packge_cost
        ,distribu_packge_cost
        ,is_bubble
        ,retrieve_status
        ,settle_franchisee_name
        ,settle_franchisee_code
        ,settle_franchisee_id
        ,is_equals_send_piece
        ,cast(to_date(bill_generation_date) as string) as dt
    from spmi_ods.spmi_union_bill where dt  = '{{ execution_date | cst_ds }}'
),
-- 最近30天数据
tmp_spmi_union_bill_incr_0030 as (
    select * from tmp_spmi_union_bill_incr where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-30) and '{{ execution_date | cst_ds }}'
),
-- 最近60-30天数据
tmp_spmi_union_bill_incr_3060 as (
    select * from tmp_spmi_union_bill_incr where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-60) and date_add('{{ execution_date | cst_ds }}',-31)
),
-- 最近90-60天数据
tmp_spmi_union_bill_incr_6090 as (
    select * from tmp_spmi_union_bill_incr where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-90) and date_add('{{ execution_date | cst_ds }}',-61)
),
-- 取dwd层90天数据
tmp_dwd_spmi_union_bill_incr_dt as (
    select
        id
        ,waybill_no
        ,settle_network_name
        ,settle_network_code
        ,settle_network_id
        ,financial_center_name
        ,financial_center_code
        ,financial_center_id
        ,waybill_num
        ,waybill_piece
        ,cost_weight
        ,settle_weight
        ,transfer_cost
        ,operation_cost
        ,piece_fee_cost
        ,distribu_depart_cost
        ,distribu_opera_cost
        ,is_package_cost
        ,product_type_id
        ,product_type_name
        ,goods_type_id
        ,goods_type_name
        ,transport_type_id
        ,transport_type_name
        ,customer_code
        ,customer_name
        ,begin_id
        ,begin_name
        ,sender_province_id
        ,sender_province_name
        ,destination_id
        ,destination_name
        ,receiver_province_id
        ,receiver_province_name
        ,waybill_time
        ,sign_time
        ,send_financial_center_id
        ,send_financial_center_name
        ,send_financial_center_code
        ,send_network_id
        ,send_network_code
        ,send_network_name
        ,sign_financial_center_id
        ,sign_financial_center_code
        ,sign_financial_center_name
        ,sign_network_id
        ,sign_network_code
        ,sign_network_name
        ,operation_network_code
        ,operation_network_name
        ,operation_financial_center_code
        ,operation_financial_center_name
        ,is_sign
        ,is_bad_piece
        ,is_refund
        ,remarks
        ,version
        ,bill_generation_date
        ,create_time
        ,update_time
        ,last_update_time_sync
        ,trace_id
        ,customer_fee
        ,distribu_version
        ,waybill_weight
        ,package_charge_weight_source
        ,center_packge_cost
        ,distribu_packge_cost
        ,is_bubble
        ,retrieve_status
        ,settle_franchisee_name
        ,settle_franchisee_code
        ,settle_franchisee_id
        ,is_equals_send_piece
        ,dt
    from spmi_dwd.dwd_spmi_union_bill_dt where dt between date_add('{{ execution_date | cst_ds }}',-90) and  '{{ execution_date | cst_ds }}'
)
-- 将ods层昨天数据与dwd层90天数据 分段去重
insert overwrite table spmi_dwd.dwd_spmi_union_bill_dt partition(dt)
select
    id
    ,waybill_no
    ,settle_network_name
    ,settle_network_code
    ,settle_network_id
    ,financial_center_name
    ,financial_center_code
    ,financial_center_id
    ,waybill_num
    ,waybill_piece
    ,cost_weight
    ,settle_weight
    ,transfer_cost
    ,operation_cost
    ,piece_fee_cost
    ,distribu_depart_cost
    ,distribu_opera_cost
    ,is_package_cost
    ,product_type_id
    ,product_type_name
    ,goods_type_id
    ,goods_type_name
    ,transport_type_id
    ,transport_type_name
    ,customer_code
    ,customer_name
    ,begin_id
    ,begin_name
    ,sender_province_id
    ,sender_province_name
    ,destination_id
    ,destination_name
    ,receiver_province_id
    ,receiver_province_name
    ,waybill_time
    ,sign_time
    ,send_financial_center_id
    ,send_financial_center_name
    ,send_financial_center_code
    ,send_network_id
    ,send_network_code
    ,send_network_name
    ,sign_financial_center_id
    ,sign_financial_center_code
    ,sign_financial_center_name
    ,sign_network_id
    ,sign_network_code
    ,sign_network_name
    ,operation_network_code
    ,operation_network_name
    ,operation_financial_center_code
    ,operation_financial_center_name
    ,is_sign
    ,is_bad_piece
    ,is_refund
    ,remarks
    ,version
    ,bill_generation_date
    ,create_time
    ,update_time
    ,last_update_time_sync
    ,trace_id
    ,customer_fee
    ,distribu_version
    ,waybill_weight
    ,package_charge_weight_source
    ,center_packge_cost
    ,distribu_packge_cost
    ,is_bubble
    ,retrieve_status
    ,settle_franchisee_name
    ,settle_franchisee_code
    ,settle_franchisee_id
    ,is_equals_send_piece
    ,dt
from (
    select
        *
    from
    (
        select
            * , row_number() over(partition by waybill_no order by  update_time desc) as rnk
        from (
            select
                *
            from tmp_dwd_spmi_union_bill_incr_dt
            where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-30) and '{{ execution_date | cst_ds }}'

            union all

            select
                *
            from tmp_spmi_union_bill_incr_0030
        ) tmp_union
    ) tmp_union_rnk
    where rnk  = 1

    union all

    select
        *
    from (
        select
            * , row_number() over(partition by waybill_no order by  update_time desc) as rnk
        from(
            select
                *
            from tmp_dwd_spmi_union_bill_incr_dt
            where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-60) and date_add('{{ execution_date | cst_ds }}',-31)

            union all

            select
                *
            from tmp_spmi_union_bill_incr_3060
        ) tmp_union
    ) tmp_union_rnk
    where rnk  = 1

    union all

    select
        *
    from (
        select
            * , row_number() over(partition by waybill_no order by  update_time desc) as rnk
        from(
            select
                *
            from tmp_dwd_spmi_union_bill_incr_dt
            where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-90) and date_add('{{ execution_date | cst_ds }}',-61)

            union all

            select
                *
            from tmp_spmi_union_bill_incr_6090
        ) tmp_union
    ) tmp_union_rnk
    where rnk  = 1
) tmp_union_all
distribute by dt, abs(hash(waybill_no)) % 100;